package com.yesep.learn.mqtt.server.broker.handler;

import java.io.IOException;

import com.yesep.learn.mqtt.protocol.IConnect;
import com.yesep.learn.mqtt.protocol.IDisConnect;
import com.yesep.learn.mqtt.protocol.IPingReq;
import com.yesep.learn.mqtt.protocol.IPubAck;
import com.yesep.learn.mqtt.protocol.IPubComp;
import com.yesep.learn.mqtt.protocol.IPubRec;
import com.yesep.learn.mqtt.protocol.IPubRel;
import com.yesep.learn.mqtt.protocol.IPublish;
import com.yesep.learn.mqtt.protocol.ISubscribe;
import com.yesep.learn.mqtt.protocol.IUnSubscribe;
import com.yesep.learn.mqtt.server.common.session.ISessionStoreService;
import com.yesep.learn.mqtt.server.common.session.SessionStore;
import com.yesep.learn.mqtt.util.ChannelUtil;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.mqtt.MqttConnectMessage;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttPubAckMessage;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.handler.codec.mqtt.MqttSubscribeMessage;
import io.netty.handler.codec.mqtt.MqttUnsubscribeMessage;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;

/**
 * @Description: MQTT消息处理
 */
public class BrokerHandler extends SimpleChannelInboundHandler<MqttMessage> {
	private IConnect connect;
	private IDisConnect disConnect;
	private ISubscribe subscribe;
	private IUnSubscribe unSubscribe;
	private IPublish publish;
	private IPubAck pubAck;
	private IPubRel pubRel;
	private IPubRec pubRec;
	private IPubComp pubComp;
	private IPingReq pingReq;
	private ISessionStoreService sessionStoreService;

	public BrokerHandler(IConnect connect, IDisConnect disConnect, ISubscribe subscribe, IUnSubscribe unSubscribe, IPublish publish, IPubAck pubAck, IPubRel pubRel, IPubRec pubRec, IPubComp pubComp, IPingReq pingReq, ISessionStoreService sessionStoreService) {
		this.connect = connect;
		this.disConnect = disConnect;
		this.subscribe = subscribe;
		this.unSubscribe = unSubscribe;
		this.publish = publish;
		this.pubAck = pubAck;
		this.pubRel = pubRel;
		this.pubRec = pubRec;
		this.pubComp = pubComp;
		this.pingReq = pingReq;
		this.sessionStoreService = sessionStoreService;
	}

	@Override
	protected void channelRead0(ChannelHandlerContext ctx, MqttMessage msg) throws Exception {
		switch (msg.fixedHeader().messageType()) {
		case CONNECT://请求连接
			connect.process(ctx.channel(), (MqttConnectMessage) msg);
			break;
		case CONNACK://请求应答
			break;
		case PUBLISH://发布消息
			publish.process(ctx.channel(), (MqttPublishMessage) msg);
			break;
		case PUBACK://发布应答
			pubAck.process(ctx.channel(), (MqttPubAckMessage) msg);
			break;
		case PUBREC://发布已接收回执
			pubRec.process(ctx.channel(), msg);
			break;
		case PUBREL://发布释放
			pubRel.process(ctx.channel(), msg);
			break;
		case PUBCOMP://发布完成
			pubComp.process(ctx.channel(), msg);
			break;
		case SUBSCRIBE://订阅请求
			subscribe.process(ctx.channel(), (MqttSubscribeMessage) msg);
			break;
		case SUBACK://订阅应答
			break;
		case UNSUBSCRIBE://取消订阅
			unSubscribe.process(ctx.channel(), (MqttUnsubscribeMessage) msg);
			break;
		case UNSUBACK://取消订阅应答
			break;
		case PINGREQ://ping请求
			pingReq.process(ctx.channel(), msg);
			break;
		case PINGRESP://ping请求回应
			break;
		case DISCONNECT://断开连接
			disConnect.process(ctx.channel(), msg);
			break;
		default:
			break;
		}
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		if (cause instanceof IOException) {
			// 远程主机强迫关闭了一个现有的连接的异常
			ctx.close();
		} else {
			super.exceptionCaught(ctx, cause);
		}
	}

	@Override
	public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
		if (evt instanceof IdleStateEvent) {
			IdleStateEvent idleStateEvent = (IdleStateEvent) evt;
			if (idleStateEvent.state() == IdleState.ALL_IDLE) {
				Channel channel = ctx.channel();
				String clientId = ChannelUtil.getClientId(channel);
				// 发送遗嘱消息
				if (sessionStoreService.containsKey(clientId)) {
					SessionStore sessionStore = sessionStoreService.get(clientId);
					if (sessionStore.getWillMessage() != null) {
						publish.process(ctx.channel(), sessionStore.getWillMessage());
					}
				}
				ctx.close();
			}
		} else {
			super.userEventTriggered(ctx, evt);
		}
	}
}
